package scala.actors.multicast.test;

import scala.actors.Actor
import scala.actors.Actor._
import scala.actors.multicast.actorgroup._
import scala.actors.multicast.handlers._

import java.net._

import edu.ucla.cs.rpc.multicast.network.MulticastManager
import edu.ucla.cs.rpc.multicast.sequencer.fixed.FixedSequencer
import edu.ucla.cs.rpc.multicast.sequencer._

/**
  This program is derived from 
  http://www.scala-lang.org/docu/examples/files/pingpong.html
  , originally written by the LAMP/EFL group.
    
  @author Micah Wendell, Chase Covello, Philip Russell
  
  The purpose of this file is to have two Actors, Ping & Pong,
  communicate through the extended (ie with multicast functionality)
  Scala Actors library.
    
  Our goal was to leave the Ping & Pong classes untouched from the original
  example, and to modify the setup/cleanup code in the runnable code segment
  ONLY.
   
  We ended up having to add another parameter to Ping for test purposes.  This
  parameter indicated the number of Pongs that would be receiving Ping's messages.
  In practice, we would not wish to rely on knowing how many receivers we are
  corresponding with, but to simply verify that the multicast mechanism is working,
  this solution is exactly what we wanted. 
  
  **/

/* 
   PingMessage defines the set of message types understood (reacted upon)
   by a Ping object.
   PongMessage defines the same behavior, but for Pong objects.
   The "case object" syntax is used so that Ping/Pong objects can case (pattern
   match) on the message types.
   */
abstract class PingMessage
case object Start extends PingMessage
case object SendPing extends PingMessage
case object Pong extends PingMessage

abstract class PongMessage
case object Ping extends PongMessage
case object Stop extends PongMessage


object PingToPong extends Application {
  /* SETUP for the MutlicastRPC framework
  	The Ping & Pong groups communicate through 2 seperate channels,
     	each comprising of a Sender, a Sequencer and a Destination, all
     	of which is managed by a MulticastManager.
     	
     	We allocate two Sequencers and two MulticastManagers
     	*/
  val manager1: MulticastManager = new MulticastManager
  val seq1: Sequencer = new FixedSequencer(manager1.getSocketAddress)
  val manager2: MulticastManager = new MulticastManager
  val seq2: Sequencer = new FixedSequencer(manager2.getSocketAddress)
  /*
    	then pass there contact information to the constructor of the ActorGroup
        that will "represent" the Ping and Pong objects to the framework
        NOTE: the Array allocation is necessary because Scala considers the type
        of the Java variable argument list to be an Array of objects. Unfortunately,
        this a small part of our implementation details.
  */
  val pingGroup: ActorGroup = new MulticastActorGroup(
     "pingGroup",
     Array[java.net.SocketAddress](seq1.getSocketAddress), 
     manager2.getSocketAddress)

  val pongGroup: ActorGroup = new MulticastActorGroup(
      "pongGroup",
      Array[java.net.SocketAddress](seq2.getSocketAddress), 
      manager1.getSocketAddress)
  /*
    Create a Ping and Pong object, and add them to their respective groups 
  */
  val pong = new Pong
  val ping = new Ping(100, 1, pingGroup)
  pingGroup += ping
  pongGroup += pong
  
  /*
    Start the Ping/Pong message bouncing behavior by sending the Start message.
  */
  ping ! Start
  
  Thread sleep 5000
  /* CLEANUP section.
    Tell the ActorGroups to Shutdown.
    If the program behaves correctly, the Ping and Pong objects should
    have terminated, sending the 'stop message to the calling environment.
    Explicitly shut down the Sequencers and MulticastManagers.
  */
  pingGroup ! Shutdown
  pongGroup ! Shutdown
  seq1.shutdown
  seq2.shutdown
  manager1.shutdown
  manager2.shutdown
}

class Ping(count: int, numpongs: int, pong: Actor) extends Actor {
  
  start
  
  def act() {
    Console.println("Ping: Initializing with count "+count+": "+pong)
    var pingsLeft = count
    //we expect one Pong message from each Pong actor for each Ping we transmit
    val pongsExpected = numpongs * count
    var pongsReceived = 0
    loop {
      react {
        case Start =>
          Console.println("Ping: starting.")
          pong ! Ping
          pingsLeft = pingsLeft - 1
        case SendPing =>
          pong ! Ping
          pingsLeft = pingsLeft - 1
        case Pong =>
          pongsReceived += 1;
          if (pingsLeft % 10 == 0) //inform console of progress
            Console.println("Ping: pong from: "+sender)
          if (pingsLeft > 0) //send another Ping by sending a SendPing to self
            self ! SendPing
          //have we received at least the correct number of pongs?
          if (pongsReceived >= pongsExpected) {
            Console.println("Ping: Stop (Received "+pongsReceived+" Pongs.)")
            pong ! Stop
            exit('stop)
          }
      }
    }
  }
}

class Pong extends Actor {
  
  start
  
  def act() {
    var pongCount = 0
    loop {
      react {
        case Ping =>
          if (pongCount % 1000 == 0)
            Console.println("Pong: ping "+pongCount+" from "+sender)
          sender ! Pong  //respond with a Pong message
          pongCount = pongCount + 1
        case Stop => //Won't receive anymore Ping messages: shutdown
          Console.println("Pong: Stop.")
          exit('stop)
        case _ => //cover the default message case
          Console.println("Pong: received unknown message")
      }
    }
  }
}
